package spark.core.scala

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Created by Administrator on 2018/1/31/031.
 */
object SparkTopN {

  /**
   * topn 求某一块数据块（表）的前几个N个值，并且把他们输出
   */
  def main(args: Array[String]) {
    //1.spark上下文
    val conf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("SprakTopN")
    val sc = new SparkContext(conf)
    //2.读取数据块，生成RDD（集合、数据集）
    val url = ("hdfs://bigdata00:8020/user/hadoop/wordcount.txt")
    val saveUrl = s"result/sparkcore_${System.currentTimeMillis()}"
    val rdd :RDD[String] = sc.textFile(url)
    //3.处理数据
    /**
     * 从每一行数据的格式中
     * 转化成（word,count）
     */
    val mapredRdd = rdd.flatMap(_.split(",")).filter(_.nonEmpty) .map(word => (word,1)).reduceByKey(_+_)

    //记得如果全局的排序，要考虑排完序之后是怎么打印的！
    //或者记得考虑将数据全部放到同一个分区当中
    //第一种方法
  //  val result:RDD[(String,Int)]= mapredRdd.sortBy(t => t._2,ascending = false,numPartitions = 1)


    //第二种方法
    val mapredRDD1: RDD[(Int, String)] =  mapredRdd.map(_.swap)
    val result: RDD[(String, Int)] = mapredRDD1.sortByKey(ascending = false,numPartitions = 1).map(_.swap)


    //另一种打印方法，但是这种打印方法，其实性能上来说，比foreachparttion差
    result.foreach(println)

    result.saveAsTextFile(saveUrl)
  }

}
